﻿using Platform.Infrastructure;
using Platform.Resources;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Linq.Expressions;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Platform.Linq
{
	/// <summary>
	/// Realiza o MapReduce
	/// </summary>
	/// <typeparam name="TInput"></typeparam>
	/// <typeparam name="TKey"></typeparam>
	/// <typeparam name="TEmit"></typeparam>
	public class MapReduceQuery<TInput, TKey, TEmit> : IMapReduceQuery<TEmit> // IQueryable<IGrouping<TKey, TEmit>> => poderia ter essa implementação também, mas utilizando somente o mapeamento não tem utilidade
	{

		/// <summary>
		/// Quantos registros serão processado em memória
		/// </summary>
		private const int RecordBufferThreshold = 3;	// TODO : Esse valor deve ser configurável, o que irá definir quantos registros é o tamanho em bytes de cada registros (quanto maior menos registros)

		/// <summary>
		/// Define o paralelismo do processo
		/// </summary>
		private const int ReduceWorkers = 3;			// TODO : Também deve ser um parâmetro, que define quantas threads serão utilizadas para fazer o mapeamento e o reduce

		/// <summary>
		/// Mantém na memória os itens que foram processados. 
		/// </summary>
		//private MapReduceBag<TKey, TEmit> bag;
		
		/// <summary>
		/// Mantém os itens que já foram processados
		/// </summary>
		private IMapReduceBag<TEmit> bag;

		/// <summary>
		/// 
		/// </summary>
		/// <param name="elements"></param>
		/// <param name="provider"></param>
		/// <param name="mapper"></param>
		//public MapReduceQuery( IQueryable<TInput> elements, IQueryProvider provider, Expression<Func<IQueryable<TInput>, IQueryable<IGrouping<TKey, TEmit>>>> mapper )
		public MapReduceQuery( IQueryable<TInput> elements, IQueryProvider provider, MapperFunction<TInput, TKey, TEmit> mapper )
		{
			this.Elements = elements;
			this.Mapper = mapper;
			this.Provider = provider;
			this.Expression = Expression.Constant( this );
		}

		/// <summary>
		/// Obtém os elementos que são utilizados como base para o processamento
		/// </summary>
		public IQueryable<TInput> Elements
		{ get; private set; }

		/// <summary>
		/// Obtém ou define a função que mapea os dados
		/// </summary>
		public MapperFunction<TInput, TKey, TEmit> Mapper
		{ get; private set; }

		/// <summary>
		/// Obtém ou define a função que realiza o processamento dos dados
		/// </summary>
		public ReducerFunction<TKey, TEmit> Reducer
		{ get; set; }

		/// <summary>
		/// Obtém ou define onde é armazenado os valores já computados
		/// </summary>
		public IMapReduceBag<TEmit> Bag
		{
			get
			{
				if ( this.bag == null )
				{
					this.bag = new MapReduceBag<TKey, TEmit>( this.Mapper.KeySelectorCompiled, this.Reducer, MapReduceKeyComparer<TKey>.Create( this.Mapper.KeySelector ) );
				}
				return this.bag;
			}
			set { this.bag = value; }
		}

		/// <summary>
		/// 
		/// </summary>
		/// <param name="index"></param>
		/// <returns></returns>
		private List<TInput> GetPartition( int index )
		{
			Debug.Assert( index >= 0 );

			IQueryable<TInput> query = this.Elements;

			if ( index > 0 )
				query = query.Skip( index * RecordBufferThreshold );
			return query.Take( RecordBufferThreshold ).ToList();
		}

		/// <summary>
		/// 
		/// </summary>
		/// <returns></returns>
		// private MapReduceBag<TKey, TEmit> Reduce()
		private void Reduce()
		{
			// TODO : Realizar teste para descobrir o tempo de processamento de milhares de itens e a quantidade de memória consumida
			// TODO : Realiza testes para descobrir o limite viável que essa coleção pode suportar
			// Mantém em memória os items que já foram processados
			//MapReduceBag<TKey, TEmit> bag = new MapReduceBag<TKey, TEmit>( this.Mapper.KeySelectorCompiled, this.Reducer, MapReduceKeyComparer<TKey>.Create( this.Mapper.KeySelector ) );


            if ( this.Bag.State == MapReduceBagState.NotInitialized )
            {
                int pageIndex = 0;
                List<TInput> partition = this.GetPartition( pageIndex );

                if ( partition.Count > 0 )
                {
                    // Objeto que faz a limitação de quantas tarefas podem ser criadas para executar ao 
                    // mesmo tempo
                    using ( SemaphoreSlim semaphore = new SemaphoreSlim( ReduceWorkers ) )
                    {
                        List<Task> tasks = new List<Task>();

                        // Se ainda há elementos para serem processados
                        while ( partition.Count > 0 )
                        {
                            //Task task = this.ReduceAsync( partition, bag, semaphore );
                            Task task = this.ReduceAsync( partition, semaphore );
                            tasks.Add( task );

                            pageIndex++;
                            partition = this.GetPartition( pageIndex );

                            // Remove as tarefas que já finalizaram
                            tasks.RemoveCompleteTasks();
                        }

                        // Aguarda finalizar as últimas tarefas
                        tasks.WaitAll();
                    }
                }
            }

			//return bag;
		}

		/// <summary>
		/// Realiza o processamento de forma assíncrona
		/// </summary>
		/// <param name="partition"></param>
		/// <param name="limiter">Controla quantas tasks serão executadas em paralelo</param>
		//private Task ReduceAsync( IEnumerable<TInput> partition, MapReduceBag<TKey, TEmit> bag, SemaphoreSlim limiter )
		private Task ReduceAsync( IEnumerable<TInput> partition, SemaphoreSlim limiter )
		{
			limiter.Wait();
			return Task.Run(
				() =>
				{
					try
					{
						var data = partition
							// Map
							.Select( this.Mapper.EmitCompiled )
							.GroupBy( this.Mapper.KeySelectorCompiled )

							// Reduce
							.Select( this.Reducer.Compiled );

						this.Bag.Merge( data );
					}
					finally
					{
						limiter.Release();
					}
				} );
		}


		public void Update( IEnumerable<TInput> freshValues )
		{
			this.Update( Enumerable.Empty<TInput>(), freshValues );
		}

		public void Update( TInput oldValues, TInput freshValues )
		{
			this.Update( new TInput[] { oldValues }, new TInput[] { freshValues } );
		}

		public void Update( IEnumerable<TInput> oldStates, IEnumerable<TInput> freshStates )
		{
            Arguments.IsNotNull( "oldStates", oldStates );
            Arguments.IsNotNull( "freshStates", freshStates );

            if ( this.Bag.State == MapReduceBagState.NotInitialized )
                throw Exceptions.MapReduceBagNotInitialized();

			// TODO: Filtrar os itens que não devem participar das contagens

			// 1. Em "oldStates", pegar a clausula "OrderBy" e obter os itens de "oldStates" que são menores que "OrderBy"
			// 2. Em "freshStates", filtrar somente os itens que são maiores que "OrderBy"


			var elements = oldStates
				// Remove os itens que ainda não foram processados
				// .Where()	

				// Map -> Mapeia para remover da lista
				.Select( this.Mapper.EmitNegatedCompiled )

				//
				.Union( 
					freshStates
						// Remove os itens que ainda não foram processados
						// .Where()
						// Map
						.Select( this.Mapper.EmitCompiled ) )

				// Reduce
				.GroupBy( this.Mapper.KeySelectorCompiled )
				.Select( this.Reducer.Compiled );

			// Realiza o merge dos dados
			this.Bag.Merge( elements );
		}

		public void Remove( TInput element )
		{
			this.Update( new TInput[] { element }, Enumerable.Empty<TInput>() );
		}

		public void Remove( IEnumerable<TInput> elements )
		{
			this.Update( elements, Enumerable.Empty<TInput>() );
		}

		#region IEnumerable<TEmit> Members

		public IEnumerator<TEmit> GetEnumerator()
		{
			// 1. Pergunta para a "Bag" se o reduce já foi processada, ou inicializado.
			//	1.1. Se sim, retorna o conteúdo da própria "Bag", senão realiza o processamento
			// 2. Realiza o processamento

			this.Reduce();
			return this.Bag.GetEnumerator();
		//	return this.Reduce().GetEnumerator();
		}

		#endregion

		#region IEnumerable Members

		System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
		{
			return this.GetEnumerator();
		}

		#endregion

		#region IQueryable Members

		public Type ElementType
		{
			get { return typeof( TEmit ); }
		}

		public Expression Expression
		{
			get;
			private set;
		}

		public IQueryProvider Provider
		{
			get;
			private set;
		}

		#endregion
	}

}
